package com.superms.demo.mq;

import java.io.Serializable;
import java.util.Collection;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.google.common.eventbus.AsyncEventBus;
import com.jeesuite.common.json.JsonUtils;
import com.jeesuite.kafka.handler.MessageHandler;
import com.jeesuite.kafka.message.DefaultMessage;
import com.jeesuite.kafka.spring.TopicProducerSpringProvider;
import com.jeesuite.kafka.thread.StandardThreadExecutor.StandardThreadFactory;
import com.jeesuite.spring.InstanceFactory;

/**
 * 
 * @description <br>
 * @author <a href="mailto:vakinge@gmail.com">vakin</a>
 * @date 2017年12月19日
 */
public class AsyncSubmitor {
	
	private static AsyncSubmitor submitor = new AsyncSubmitor();

	private AsyncEventBus asyncEventBus;
	
	private TopicProducerSpringProvider topicProducer;
	
	private ExecutorService executorService;
	
	public static AsyncSubmitor getSubmitor() {
		return submitor;
	}

	private AsyncSubmitor() {
		try {			
			topicProducer = InstanceFactory.getInstance(TopicProducerSpringProvider.class);
		} catch (Exception e) {}
		
		if(topicProducer != null)return;
		
		executorService = Executors.newFixedThreadPool(2,new StandardThreadFactory("async-event-executor"));
		asyncEventBus = new AsyncEventBus(executorService);
		Collection<MessageHandler> handlers = InstanceFactory.getInstanceProvider().getInterfaces(MessageHandler.class).values();
		for (MessageHandler messageHandler : handlers) {
			asyncEventBus.register(messageHandler);
		}
		
	}
	
	public void submit(String topic,Object message){
		if(topicProducer != null){
			if(message instanceof DefaultMessage){
				topicProducer.publish(topic, (DefaultMessage)message);
			}else{
				topicProducer.publishNoWrapperMessage(topic, JsonUtils.toJson(message));
			}
			
		}else{
			asyncEventBus.post(message);
		}
	}
	
	public void close(){
		if(executorService != null){
			executorService.shutdown();
			executorService = null;
		}
	}
}
